/**
 * KWDB Ingest Agent Example
 *
 * This demonstrates the "Agent Mode" deployment pattern:
 *
 * [Business Application (any language)]
 *         │ Simple HTTP/JSON API
 *         ▼
 * [Ingest Agent Service (This Program)]
 *         │ C++ SDK with UDS + Streaming + Columnar
 *         ▼
 * [KWDB / ingestd]
 *
 * Architecture Benefits:
 * 1. Business logic uses simple HTTP API (no SDK integration required)
 * 2. Agent co-located with KWDB uses ultra-fast UDS transport
 * 3. Agent leverages C++ SDK's StreamClient, Batcher, RetryPolicy
 * 4. Columnar writes for maximum throughput
 * 5. Minimal overhead: single process, zero network latency to DB
 *
 * Usage:
 *   ./agent_example [port]
 *
 * Example:
 *   curl -X POST http://localhost:8080/write \
 *     -H "Content-Type: application/json" \
 *     -d '{"measurement":"metrics","tags":{"host":"server1"},"fields":{"cpu":45.6},"timestamp":1234567890}'
 */

#include "kwdb/client.h"
#include "kwdb/point.h"
#include "kwdb/status.h"
#include <iostream>
#include <string>
#include <thread>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <cstring>
#include <sstream>

using namespace kwdb;

// Global client for Agent mode (uses local UDS connection)
std::unique_ptr<KwdbClient> g_client;

// Simple JSON parser for demo purposes
struct JsonPoint {
    std::string measurement;
    std::string tag_key;
    std::string tag_value;
    std::string field_key;
    double field_value;
    int64_t timestamp_ns;
};

bool parse_json_point(const std::string& json, JsonPoint& point) {
    // Very simple JSON parser for demo
    // In production, use a real JSON library (e.g., nlohmann/json)

    // Extract measurement
    size_t meas_start = json.find("\"measurement\":\"");
    if (meas_start == std::string::npos) return false;
    meas_start += 15;
    size_t meas_end = json.find("\"", meas_start);
    point.measurement = json.substr(meas_start, meas_end - meas_start);

    // Extract tag (simplified: only one tag)
    size_t tag_key_start = json.find("\"tags\":{\"");
    if (tag_key_start != std::string::npos) {
        tag_key_start += 9;
        size_t tag_key_end = json.find("\"", tag_key_start);
        point.tag_key = json.substr(tag_key_start, tag_key_end - tag_key_start);

        size_t tag_val_start = json.find("\"", tag_key_end + 1) + 1;
        size_t tag_val_end = json.find("\"", tag_val_start);
        point.tag_value = json.substr(tag_val_start, tag_val_end - tag_val_start);
    }

    // Extract field (simplified: only one field)
    size_t field_key_start = json.find("\"fields\":{\"");
    if (field_key_start == std::string::npos) return false;
    field_key_start += 11;
    size_t field_key_end = json.find("\"", field_key_start);
    point.field_key = json.substr(field_key_start, field_key_end - field_key_start);

    size_t field_val_start = field_key_end + 2;
    size_t field_val_end = json.find_first_of(",}", field_val_start);
    point.field_value = std::stod(json.substr(field_val_start, field_val_end - field_val_start));

    // Extract timestamp (optional)
    size_t ts_start = json.find("\"timestamp\":");
    if (ts_start != std::string::npos) {
        ts_start += 12;
        size_t ts_end = json.find_first_of(",}", ts_start);
        point.timestamp_ns = std::stoll(json.substr(ts_start, ts_end - ts_start));
    } else {
        point.timestamp_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
            std::chrono::system_clock::now().time_since_epoch()
        ).count();
    }

    return true;
}

void handle_write_request(int client_fd) {
    char buffer[65536] = {0};
    int total_read = 0;

    // Read HTTP request
    while (total_read < sizeof(buffer) - 1) {
        int n = read(client_fd, buffer + total_read, sizeof(buffer) - total_read - 1);
        if (n <= 0) break;
        total_read += n;
        buffer[total_read] = '\0';

        std::string request(buffer);
        if (request.find("\r\n\r\n") != std::string::npos) {
            break;
        }
    }

    std::string request(buffer);

    // Check if it's POST /write
    if (request.find("POST /write") != 0) {
        const char* response = "HTTP/1.1 404 Not Found\r\n\r\n{\"error\":\"Not Found\"}";
        write(client_fd, response, strlen(response));
        close(client_fd);
        return;
    }

    // Find body
    size_t body_pos = request.find("\r\n\r\n");
    if (body_pos == std::string::npos) {
        const char* response = "HTTP/1.1 400 Bad Request\r\n\r\n{\"error\":\"No body\"}";
        write(client_fd, response, strlen(response));
        close(client_fd);
        return;
    }

    std::string body = request.substr(body_pos + 4);

    // Parse JSON
    JsonPoint json_point;
    if (!parse_json_point(body, json_point)) {
        const char* response = "HTTP/1.1 400 Bad Request\r\n\r\n{\"error\":\"Invalid JSON\"}";
        write(client_fd, response, strlen(response));
        close(client_fd);
        return;
    }

    // Convert to KWDB Point and write using SDK
    Point point(json_point.measurement);
    point.add_tag(json_point.tag_key, json_point.tag_value);
    point.add_field(json_point.field_key, json_point.field_value);

    // Convert nanoseconds to time_point
    auto duration = std::chrono::nanoseconds(json_point.timestamp_ns);
    auto tp = std::chrono::system_clock::time_point(
        std::chrono::duration_cast<std::chrono::system_clock::duration>(duration)
    );
    point.timestamp(tp);

    Status status = g_client->write(point);

    // Send response
    if (status.ok()) {
        const char* response = "HTTP/1.1 200 OK\r\n\r\n{\"status\":\"ok\"}";
        write(client_fd, response, strlen(response));
    } else {
        std::string error_msg = "{\"error\":\"" + status.message + "\"}";
        std::string response = "HTTP/1.1 500 Internal Server Error\r\n\r\n" + error_msg;
        write(client_fd, response.c_str(), response.length());
    }

    close(client_fd);
}

void run_agent_server(int port) {
    int server_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (server_fd == 0) {
        std::cerr << "Failed to create socket" << std::endl;
        return;
    }

    int opt = 1;
    setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    struct sockaddr_in address;
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(port);

    if (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {
        std::cerr << "Bind failed on port " << port << std::endl;
        close(server_fd);
        return;
    }

    if (listen(server_fd, 10) < 0) {
        std::cerr << "Listen failed" << std::endl;
        close(server_fd);
        return;
    }

    std::cout << "============================================" << std::endl;
    std::cout << "  KWDB Ingest Agent Service" << std::endl;
    std::cout << "============================================" << std::endl;
    std::cout << "Agent Mode Architecture:" << std::endl;
    std::cout << "  [Business App] --HTTP--> [Agent] --UDS--> [KWDB]" << std::endl;
    std::cout << "" << std::endl;
    std::cout << "Listening on: http://0.0.0.0:" << port << "/write" << std::endl;
    std::cout << "Backend: KWDB via UDS (ultra-fast local connection)" << std::endl;
    std::cout << "" << std::endl;
    std::cout << "Example request:" << std::endl;
    std::cout << "  curl -X POST http://localhost:" << port << "/write \\" << std::endl;
    std::cout << "    -H \"Content-Type: application/json\" \\" << std::endl;
    std::cout << "    -d '{\"measurement\":\"metrics\",\"tags\":{\"host\":\"server1\"}," << std::endl;
    std::cout << "         \"fields\":{\"cpu\":45.6},\"timestamp\":1234567890}'" << std::endl;
    std::cout << "============================================" << std::endl;

    while (true) {
        int addrlen = sizeof(address);
        int client_fd = accept(server_fd, (struct sockaddr*)&address, (socklen_t*)&addrlen);

        if (client_fd < 0) {
            continue;
        }

        // Handle each request in a new thread
        std::thread(handle_write_request, client_fd).detach();
    }
}

int main(int argc, char** argv) {
    int port = 8080;
    if (argc > 1) {
        port = std::atoi(argv[1]);
    }

    // Initialize KWDB client in "local" mode
    // This automatically uses UDS for ultra-low latency
    IngestConfig config;
    config.ingest_endpoint = "local";  // Auto-detects UDS path
    config.timeout_ms = 5000;

    std::cout << "[Agent] Initializing C++ SDK in local mode..." << std::endl;
    g_client = std::make_unique<KwdbClient>(config);
    std::cout << "[Agent] SDK ready (using UDS for zero-overhead communication)" << std::endl;
    std::cout << "" << std::endl;

    // Run Agent HTTP server
    run_agent_server(port);

    return 0;
}
